package com.qingcloud.df.executor.launcher;

import cn.hutool.core.collection.CollUtil;
import com.qingcloud.df.executor.queue.subscriber.ComponentSubscriber;
import com.qingcloud.df.executor.utils.ComponentUtil;
import com.qingcloud.df.sdk.component.config.CompInstBO;
import com.qingcloud.df.sdk.component.context.CompInstContext;
import com.qingcloud.df.sdk.component.transport.akka.AkkaTransporter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;


/**
 * 消费数据类型组件的启动器
 *
 * @author wangjinnan
 */

public class ConsumerLauncher implements  Launcher {

    Logger logger = LoggerFactory.getLogger(ConsumerLauncher.class);
    @Override
    public void start(CompInstContext context) {
        List<CompInstBO> parentInstanceList = context.getParentInstanceList();
        if (CollUtil.isNotEmpty(parentInstanceList)) {
            List<String> topics = new ArrayList<>();
            for (CompInstBO componentInstanceBO : parentInstanceList) {
                topics.add(componentInstanceBO.getTopic());
            }
            try {
                ComponentSubscriber  componentSubscriber = ComponentUtil.getComponentSubscriber(context);
                componentSubscriber.subscribe(context, topics);
            } catch (Exception e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        }
    }
    /**
     * 停止组件消费线程方法
     */
    @Override
    public void stop(CompInstContext context) {
        try {
            List<CompInstBO> parentInstanceList = context.getParentInstanceList();
            if (CollUtil.isNotEmpty(parentInstanceList)) {
                List<String> topics = new ArrayList<>();
                for (CompInstBO componentInstanceBO : parentInstanceList) {
                    topics.add(componentInstanceBO.getTopic());
                }
                ComponentSubscriber  componentSubscriber = ComponentUtil.getComponentSubscriber(context);
                componentSubscriber.unsubscribe(context,topics);
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
}
